package cn.zhage.aimin.canal.client.client;

import cn.zhage.aimin.canal.client.config.CanalProperties;
import cn.zhage.aimin.canal.client.handler.MessageHandler;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.Message;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Scheduled;

import java.net.InetSocketAddress;
import java.util.List;
import java.util.stream.Stream;

/*
 *
 * @author 渣哥
 */
@Slf4j
public abstract class AbstractCanalClient implements CanalClient {

    CanalConnector connector;
    final CanalProperties canalProperties;
    private final MessageHandler<Message> messageHandler;

    public AbstractCanalClient(CanalProperties canalProperties, MessageHandler<Message> messageHandler){
        this.canalProperties = canalProperties;
        this.messageHandler = messageHandler;
    }

    @Override
    @PostConstruct
    public void connect() {
        connector = createConnector();
        connector.connect();
        connector.subscribe(canalProperties.getSubscribe());
        connector.rollback();
    }


    @Override
    @Scheduled(initialDelayString = "${canal.initialDelay:2000}", fixedDelayString = "${canal.fixedDelay:2000}")
    public void receive() {
        log.info("canal-client 获取数据...");
        Message message = connector.getWithoutAck(canalProperties.getBatchSize());
        messageHandler.handle(message);
        connector.ack(message.getId());
    }

    @Override
    @PreDestroy
    public void disconnect() {
        log.info("释放CanalConnector successfully!");
        connector.disconnect();
    }


    public List<InetSocketAddress> formatAddress() {
        try{
            String address = canalProperties.getAddress();
            return Stream.of(address.split(",")).map(item ->{
                String[] split = item.split(":");
                return new InetSocketAddress(split[0], Integer.parseInt(split[1]));
            }).toList();
        }catch (Exception e){
            new RuntimeException("canal address format error");
        }
        return null;
    }
}
